package com.atguigu.day08;

import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

//通过一条的不同的数据 来控制另一条流的逻辑
public class Flink07_OperatorState_Broadcast {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.分别获取两条流
        DataStreamSource<String> localStream = env.socketTextStream("localhost", 9999);
        DataStreamSource<String> hadoopStream = env.socketTextStream("hadoop102", 9999);

        //3.通过local流里面的数据控制hadoop流的逻辑 因此将local流里面的内容广播出去
        MapStateDescriptor<String, String> mapState = new MapStateDescriptor<>("mapState", String.class, String.class);
        BroadcastStream<String> broadcast = localStream.broadcast(mapState);

        //4.连接两条流
        BroadcastConnectedStream<String, String> connect = hadoopStream.connect(broadcast);

        //5.使用process处理两条流
        connect.process(new BroadcastProcessFunction<String, String, String>() {
            @Override
            public void processElement(String value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
                //获取广播状态
                ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapState);
                //提取状态中的数据进而根据数据来判断执行的逻辑
                String aSwitch = broadcastState.get("switch");
                if ("a".equals(aSwitch)) {
                    out.collect("执行逻辑1........");
                } else if ("b".equals(aSwitch)) {
                    out.collect("执行逻辑2........");
                } else {
                    out.collect("执行其他逻辑........");
                }

            }

            @Override
            public void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {
                //获取广播状态
                BroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapState);
                //将数据写入状态
                broadcastState.put("switch", value);
            }
        }).print();

        env.execute();
    }
}
